package com.sky.business.sys.util;

import com.sky.business.sys.domain.SystemConfig;
import com.sky.business.sys.mqtt.listener.CMCCDispatcherCallbackListener;
import com.sky.business.sys.mqtt.listener.LocalReportingCallbackListener;
import com.sky.business.sys.service.SystemConfigService;
import com.sky.common.config.CacheConstants;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;

import java.nio.charset.StandardCharsets;

/**
 * 控制中心与移动通信的MQTT工具类
 */
@Slf4j
@Order(2)
@Configuration
public class MQTTClientUtils {

    /**
     * 连接移动MQTT客户端
     */
    public static MqttClient CMCC_MQTT_CLIENT;

    /**
     * 连接本地MQTT客户端
     */
    public static MqttClient LOCAL_MQTT_CLIENT;


    /**
     * 本地心跳上报Topic
     */
    public static String LOCAL_REPORTING_TOPIC;

    /**
     * 移动任务下发Topic
     */
    public static String CMCC_DISPATCHER_TOPIC;

    /**
     * 移动上报心跳Topic
     */
    public static String CMCC_REPORTING_TOPIC;

    @Resource
    private CMCCDispatcherCallbackListener dispatcherCallbackListener;

    @Resource
    private LocalReportingCallbackListener localReportingCallbackListener;

    @Resource
    private SystemConfigService systemConfigService;


    /**
     * 创建本地MQTT客户端方法
     * @param host 主机
     * @param port 端口
     * @param clientId 客户端id
     * @param options 连接选项
     * @param localReportingCallbackListener 回调方法
     */
    public static void createLocalMQTTClient(String host, String port, String clientId, MqttConnectOptions options,LocalReportingCallbackListener localReportingCallbackListener) throws MqttException {
        try{
            LOCAL_MQTT_CLIENT = new MqttClient("tcp://"+host+":"+port, clientId+"_"+System.currentTimeMillis());
            LOCAL_MQTT_CLIENT.setTimeToWait(30000);
            LOCAL_MQTT_CLIENT.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    try {
                        // 重连后重新订阅
                        log.info("本地MQTT{}连接成功",reconnect?"重新":"初次");
                        subscribe(MQTTClientUtils.LOCAL_MQTT_CLIENT, MQTTClientUtils.LOCAL_REPORTING_TOPIC, CacheConstants.LOCAL_SUBSCRIBE_QOS, localReportingCallbackListener);
                        log.info("本地MQTT订阅成功");
                    } catch (MqttException e) {
                        log.error("本地MQTT订阅发生异常",e);
                    }
                }

                @Override
                public void connectionLost(Throwable cause) {
                    log.error("本地MQTT连接丢失",cause);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                }
            });
            connect(LOCAL_MQTT_CLIENT, options);
        }catch (MqttException exception){
            log.error("本地MQTT连接失败！,ReasonCode:{}", exception.getReasonCode());
            throw exception;
        }
    }

    /**
     * 创建移动MQTT客户端方法
     * @param host 主机
     * @param port 端口
     * @param clientId 客户端id
     * @param options 选项
     * @param dispatcherCallbackListener 回调对象
     */
    public static void createCMCCMQTTClient(String host, String port, String clientId, MqttConnectOptions options,CMCCDispatcherCallbackListener dispatcherCallbackListener) throws MqttException {
        try{
            CMCC_MQTT_CLIENT = new MqttClient("tcp://"+host+":"+port, clientId+"_"+System.currentTimeMillis());
            CMCC_MQTT_CLIENT.setTimeToWait(30000);
            CMCC_MQTT_CLIENT.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    try {
                        // 重连后重新订阅
                        log.info("云边MQTT{}连接成功",reconnect?"重新":"初次");
                        subscribe(MQTTClientUtils.CMCC_MQTT_CLIENT, MQTTClientUtils.CMCC_DISPATCHER_TOPIC, CacheConstants.CMCC_SUBSCRIBE_QOS, dispatcherCallbackListener);
                        log.info("云边MQTT订阅成功");
                    } catch (MqttException e) {
                        log.error("云边MQTT订阅发生异常",e);
                    }
                }

                @Override
                public void connectionLost(Throwable cause) {
                    log.error("移动凌云平台MQTT连接丢失",cause);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                }
            });
            connect(CMCC_MQTT_CLIENT, options);
        }catch (MqttException exception){
            log.error("移动凌云平台MQTT连接失败！ReasonCode:{}",exception.getReasonCode());
            throw exception;
        }
    }

    /**
     * 连接mqtt服务端
     * @param client 客户端
     * @param options 选项
     * @throws MqttException mqtt异常
     */
    public static void connect(MqttClient client, MqttConnectOptions options) throws MqttException {
        client.connect(options);
    }


    /**
     * 消息发送 : retained 默认为 false
     * "retained message" 指的是 Broker 会保留的最后一条发布到某个主题的消息。
     * 当新的订阅者连接到该主题时，Broker 会将这条保留消息立即发送给订阅者，即使在订阅者订阅时该消息并未被重新发布。
     * 这对于一些需要初始状态或者最后一次已知状态的应用场景非常有用。
     * @param topicName 主题名称
     * @param message 消息
     * @param qos  服务质量
     * @return 是否成功
     */
    public static boolean publish(MqttClient client, String topicName, int qos, String message) {
        log.debug("开始发送消息，主题名:{}, qos:{}, 消息:{}", topicName, qos, message);
        MqttMessage mqttMessage = new MqttMessage(message.getBytes(StandardCharsets.UTF_8));
        mqttMessage.setQos(qos);
        mqttMessage.setRetained(false);
        try {
            client.publish(topicName, mqttMessage);
            return true;
        }catch (MqttException exception){
            log.error("消息发送失败,ReasonCode:{}", exception.getReasonCode());
            exception.printStackTrace();
            return false;
        }
    }

    /**
     * 订阅某个主题
     *
     * @param topicName 主题名称
     * @param qos 服务质量
     */
    public static void subscribe(MqttClient client ,String topicName, int qos, IMqttMessageListener messageListener) throws MqttException {
        log.info("订阅主题名:{}, qos:{}, Listener类:{}", topicName, qos, messageListener.getClass());
        try {
            client.subscribe(topicName, qos, messageListener);
        } catch (MqttException exception) {
            log.error("订阅主题失败,ReasonCode:{}", exception.getReasonCode());
            throw exception;
        }
    }

    /**
     * 取消订阅主题
     * @param topicName 主题名称
     */
    public static void cleanTopic(MqttClient client, String topicName) throws MqttException {
        log.info("取消订阅主题名:{}", topicName);
        try {
            client.unsubscribe(topicName);
        } catch (MqttException exception) {
            log.error("取消订阅出错,ReasonCode:{}", exception.getReasonCode());
            throw exception;
        }
    }

    /**
     * 销毁CMCC的MQTT连接客户端
     * @param client 客户端
     */
    public static void disconnect(MqttClient client) throws MqttException {
        try {
            client.disconnect();
            log.info("已断开MQTT连接");
        } catch (MqttException exception) {
            log.error("断开MQTT连接失败,ReasonCode:{}", exception.getReasonCode());
            throw exception;
        }
    }

    /**
     * 构建MQTTClient初始化时需要的参数
     * @param username 用户名
     * @param password 密码
     * @return 连接参数
     */
    public static MqttConnectOptions createOptions(String username, String password){
        //构建用于MQTT的初始化客户端参数
        MqttConnectOptions options = new MqttConnectOptions();
        // 用户名
        options.setUserName(username);
        // 密码
        options.setPassword(password.toCharArray());
        // 保留会话，支持恢复订阅
        options.setCleanSession(false);
        // 启用自动重连
        options.setAutomaticReconnect(true);
        // 连接超时时间
        options.setConnectionTimeout(10);
        //允许同时存在的未完成 QoS 1和QoS 2消息的最大数量
        options.setMaxInflight(100);
        //心跳间隔时间
        options.setKeepAliveInterval(20);
        return options;
    }




    /**
     * 初始化方法
     */
    @PostConstruct
    public void initMqttClient(){

        //从数据库获取原始配置
        SystemConfig config = systemConfigService.getSystemConfig();

        //移动和本地的任务下发和心跳上报的Topic保存在工具类的静态属性中
        CMCC_DISPATCHER_TOPIC = config.getCmccDispatcherTopic();
        CMCC_REPORTING_TOPIC = config.getCmccReportingTopic();
        LOCAL_REPORTING_TOPIC = config.getLocalReportingTopic();

        //构建用于移动MQTT的初始化客户端参数
        MqttConnectOptions cmccOptions = createOptions(config.getCmccMQTTServiceUsername(), config.getCmccMQTTServicePassword());

        //构建用于本地MQTT的初始化客户端参数
        MqttConnectOptions localOptions = createOptions(config.getLocalMQTTServiceUsername(), config.getLocalMQTTServicePassword());

        try {
            //创建移动MQTT连接
            createCMCCMQTTClient(config.getCmccMQTTServiceHost(), config.getCmccMQTTServicePort(), config.getCmccMQTTClientId(), cmccOptions,dispatcherCallbackListener);
        }catch (Exception e){
            log.error("移动MQTT初始化连接发生异常",e);
        }

        try {
            //创建本地的MQTT连接
            createLocalMQTTClient(config.getLocalMQTTServiceHost(), config.getLocalMQTTServicePort(), config.getLocalMQTTClientId(), localOptions,localReportingCallbackListener);
        }catch (Exception e){
            log.error("本地MQTT初始化连接发生异常",e);
        }
    }
}

